package _50_rocketmq._consumer01;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 消费者消费消息步骤
 *      创建一个消费者
 *          使用默认类推模式，初始化一个消费者示例
 *              推模式：broker 收到消息后，主动给 消费者推送，即push
 *              拉模式：消费者 每次自己查询 broker 是否有消息可以消费，即pull
 *          配置必要参数：
 *              消费者组，topic, nameserver
 *              消费位置（ConsumeFromWhere）
 *                  .CONSUME_FROM_FIRST_OFFSET      从第一个消息开始消费
 *                  .CONSUME_FROM_LAST_OFFSET       从上次位置继续消费
 *                  .CONSUME_FROM_TIMESTAMP         从指定时间开始
 *      开启消费者服务
 */
public class ConsumerTest {
    public static final String CONSUMER_GROUP = "test_aa_consumer_group_name";
    public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";
    public static final String TOPIC = "TopicTest";
    public static void main(String[] args) throws MQClientException {
        // 初始化一个 Push 类型的 消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        // 可以不指定，则读取环境变量
        consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
        // 消费哪个 topic, * 是所有信息都不过滤 -- 过滤机制
        consumer.subscribe(TOPIC, "*");
        // 配置从哪个位置开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

//        consumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//                return null;
//            }
//        });

        // 在 push 模式下，注册监听到消息后 业务逻辑（broker推送消息后）
        consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {
            //
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        // 开启 消费者服务
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
